package org.raftkv.raft.quorum;

import org.raftkv.commons.Tuple;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Comparator;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * MajorityConfig 是一组 ID，用于多数仲裁来做出决策。
 *
 * @author wzy
 */
public class MajorityConfig extends HashSet<Long> implements Serializable {

    public MajorityConfig() {
    }

    public MajorityConfig(Set<Long> ids) {
        if (ids == null) {
            return;
        }
        this.addAll(ids);
    }

    @Override
    public String toString() {
        List<Long> sl = new ArrayList<>(this);
        sl.sort((a, b) -> Long.compare(b, a));
        StringBuilder buf = new StringBuilder();
        buf.append("(");
        for (int i = 0; i < sl.size(); i++) {
            if (i > 0) {
                buf.append(" ");
            }
            buf.append(sl.get(i));
        }
        buf.append(")");
        return buf.toString();
    }

    /**
     * 该函数返回给定查找器的提交索引的（多行）表示形式。
     */
    public String describe(AckedIndexer l) {
        if (this.size() == 0) {
            return "<empty majority quorum>";
        }

        class Tup {

            long id;

            long idx;

            boolean ok;

            int bar;

            public Tup(long id, long idx, boolean ok) {
                this.id = id;
                this.idx = idx;
                this.ok = ok;
            }

        }

        int n = this.size();
        List<Tup> info = new ArrayList<>();

        for (long id : this) {
            Tuple<Long, Boolean> t = l.ackedIndex(id);
            Long idx = t.getFirst();
            Boolean ok = t.getSecond();
            info.add(new Tup(id, idx, ok));
        }

        info.sort((a, b) -> {
            if (a.idx == b.idx) {
                return Long.compare(a.id, b.id);
            }
            return Long.compare(a.idx, b.idx);
        });

        for (int i = 0; i < info.size(); i++) {
            if (i > 0 && info.get(i - 1).idx < info.get(i).idx) {
                info.get(i).bar = i;
            }
        }

        info.sort(Comparator.comparingLong(a -> a.id));

        StringBuilder buf = new StringBuilder();
        buf.append(" ".repeat(n));
        buf.append("    idx\n");
        for (Tup tup : info) {
            int bar = tup.bar;

            if (!tup.ok) {
                buf.append("?").append(" ".repeat(n));
            } else {
                buf.append("x".repeat(bar)).append(">").append(" ".repeat(n - bar));
            }
            buf.append(String.format(" %5d    (id=%d)\n", tup.idx, tup.id));
        }

        return buf.toString();
    }

    /**
     * 该函数把一个 MajorityConfig 以 List 的形式返回。
     */
    public List<Long> asList() {
        List<Long> sl = new ArrayList<>(this);
        sl.sort(Long::compare);

        return sl;
    }

    /**
     * 该函数根据通过提供的 AckedIndexer 提供的索引（对于活动配置）计算提交的索引。
     */
    public long committedIndex(AckedIndexer l) {
        int n = this.size();
        if (n == 0) {
            return Long.MAX_VALUE;
        }

        Long[] srt = new Long[n];
        int i = n - 1;
        for (Long id : this) {
            Tuple<Long, Boolean> t = l.ackedIndex(id);
            Long idx = t.getFirst();
            Boolean ok = t.getSecond();
            if (ok) {
                srt[i] = idx;
                i--;
            }
        }

        Arrays.sort(srt, Long::compare);

        int pos = n - (n / 2 + 1);
        return srt[pos];
    }

    /**
     * 该函数将选民及其投票映射到一个 Map 中，然后返回一个结果，该结果指示本轮投票是 Pending、Won 还是 Lost。
     *
     * @see VoteResult
     */
    public VoteResult voteResult(Map<Long, Boolean> votes) {
        // 按照惯例，空配置的选举获胜
        // 这对于联合仲裁非常有用，因为它会使半填充的联合仲裁的行为类似于多数仲裁
        if (this.size() == 0) {
            return VoteResult.VoteWon;
        }

        // 设置一个数组分别统计 yes 和 no 的选票数量
        int[] ny = new int[2];

        int missing = 0;
        for (Long id : this) {
            Boolean v = votes.get(id);
            if (v == null) {
                missing++;
                continue;
            }

            if (v) {
                ny[1]++;
            } else {
                ny[0]++;
            }
        }

        int q = this.size() / 2 + 1;
        if (ny[1] >= q) {
            return VoteResult.VoteWon;
        }
        if (ny[1] + missing >= q) {
            return VoteResult.VotePending;
        }

        return VoteResult.VoteLost;
    }

}
